/**
 * 
 */
package org.wolf.cluster.server;

import java.io.BufferedReader;
import java.io.File;
import java.io.FileReader;
import java.io.IOException;
import java.net.InetAddress;
import java.net.InetSocketAddress;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Set;

import org.apache.commons.cli.BasicParser;
import org.apache.commons.cli.CommandLine;
import org.apache.commons.cli.HelpFormatter;
import org.apache.commons.cli.Options;
import org.apache.commons.cli.ParseException;
import org.apache.zookeeper.server.NIOServerCnxn;
import org.apache.zookeeper.server.ZKDatabase;
import org.apache.zookeeper.server.ZooKeeperServer;
import org.apache.zookeeper.server.persistence.FileTxnSnapLog;
import org.apache.zookeeper.server.quorum.QuorumPeer;
import org.apache.zookeeper.server.quorum.QuorumPeer.LearnerType;
import org.apache.zookeeper.server.quorum.QuorumPeer.QuorumServer;
import org.apache.zookeeper.server.quorum.flexible.QuorumHierarchical;
import org.apache.zookeeper.server.quorum.flexible.QuorumMaj;
import org.apache.zookeeper.server.quorum.flexible.QuorumVerifier;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.wolf.common.config.ClusterCfg;
import org.wolf.common.config.ClusterServerCfg;
import org.wolf.common.config.WolfCfg;
import org.wolf.common.exception.ValidationException;
import org.wolf.common.util.ClassUtil;
import org.wolf.common.util.Constants;
import org.wolf.common.util.NetUtil;
import org.wolf.common.util.XMLUtil;

/**
 * @author 何小锋
 * 
 */
public class ClusterServer implements Runnable {
	/**
	 * Logger for this class
	 */
	private static final Logger LOG = LoggerFactory
			.getLogger(ClusterServer.class);

	private String config;

	private Integer port;

	private String dataDir;

	public ClusterServer(String config, Integer port, String dataDir) {
		super();
		this.config = config;
		this.port = port;
		this.dataDir = dataDir;
	}

	public void run() {
		File file = config == null ? ClassUtil
				.getResourceAsFile(Constants.DEFAULT_CONFIG) : new File(config);
		if (file == null || !file.exists()) {
			LOG.error("config file is not found");
			return;
		}
		try {
			WolfCfg wolfCfg = XMLUtil.unmarshall(WolfCfg.class, file);
			ClusterCfg clusterCfg = wolfCfg.getCluster() == null ? new ClusterCfg()
					: wolfCfg.getCluster();
			if (port != null)
				clusterCfg.setClientPort(port);
			if (dataDir != null)
				clusterCfg.setDataDir(dataDir);

			InetSocketAddress socketAddress = null;
			if (clusterCfg.getClientPortAddress() != null) {
				socketAddress = new InetSocketAddress(
						InetAddress
								.getByName(clusterCfg.getClientPortAddress()),
						clusterCfg.getClientPort());
			} else {
				socketAddress = new InetSocketAddress(
						clusterCfg.getClientPort());
			}

			// 数据目录
			File dataDir = dirExists(clusterCfg.getDataDir());
			// 日志目录
			dirExists(clusterCfg.getDataLogDir());
			// myid文件
			File idFile = fileExists(dataDir, "myid");
			Long myId = readMyId(idFile);
			FileTxnSnapLog ftxn = new FileTxnSnapLog(new File(
					clusterCfg.getDataLogDir()), new File(
					clusterCfg.getDataDir()));
			NIOServerCnxn.Factory cnxnFactory = new NIOServerCnxn.Factory(
					socketAddress, clusterCfg.getMaxClientCnxns());

			Set<String> ips = NetUtil.getAllAddress();
			HashMap<Long, QuorumServer> servers = new HashMap<Long, QuorumServer>();
			HashMap<Long, QuorumServer> obervers = new HashMap<Long, QuorumServer>();
			HashMap<Long, Long> serverGroup = new HashMap<Long, Long>();
			HashMap<Long, Long> serverWeight = new HashMap<Long, Long>();
			Set<Long> groups = new HashSet<Long>();
			if (clusterCfg.getServers() != null) {
				for (int i = 0; i < clusterCfg.getServers().size(); i++) {
					Long id = Long.valueOf(i);
					ClusterServerCfg serverCfg = clusterCfg.getServers().get(i);
					if (serverCfg.getIp() == null
							|| serverCfg.getIp().isEmpty()) {
						throw new ValidationException("server(" + i
								+ ") ip is empty");
					}
					if (myId == null && ips.contains(serverCfg.getIp())) {
						myId = id;
					}
					LearnerType learnerType = getLearnerType(serverCfg
							.getType());
					QuorumServer quorumServer = new QuorumServer(id,
							new InetSocketAddress(serverCfg.getIp(), serverCfg
									.getPort()),
							new InetSocketAddress(serverCfg.getIp(), serverCfg
									.getElectionPort()), learnerType);
					if (learnerType == LearnerType.PARTICIPANT) {
						servers.put(id, quorumServer);
					} else {
						obervers.put(id, quorumServer);
					}
					if (serverCfg.getGroup() != null) {
						serverGroup.put(id, serverCfg.getGroup());
						groups.add(serverCfg.getGroup());
					}
					if (serverCfg.getWeight() != null)
						serverWeight.put(id, serverCfg.getWeight());
				}
			}
			if (servers.size() <= 1) {
				// 单机模式
				ZooKeeperServer zkServer = new ZooKeeperServer();
				zkServer.setTxnLogFactory(ftxn);
				zkServer.setTickTime(clusterCfg.getTickTime());
				zkServer.setMinSessionTimeout(clusterCfg.getMinSessionTimeout());
				zkServer.setMaxSessionTimeout(clusterCfg.getMaxSessionTimeout());
				cnxnFactory.startup(zkServer);
				cnxnFactory.join();
				if (zkServer.isRunning()) {
					zkServer.shutdown();
				}
			} else {
				// 集群模式
				servers.putAll(obervers);
				QuorumVerifier quorumVerifier = servers.isEmpty() ? null
						: serverGroup.isEmpty() ? new QuorumMaj(servers.size())
								: new QuorumHierarchical(groups.size(),
										serverWeight, serverGroup);
				QuorumPeer quorumPeer = new QuorumPeer();
				quorumPeer.setClientPortAddress(socketAddress);
				quorumPeer.setTxnFactory(ftxn);

				quorumPeer.setQuorumPeers(servers);
				quorumPeer.setElectionType(clusterCfg.getElectionAlg());
				quorumPeer.setMyid(myId == null ? 0 : myId);
				quorumPeer.setTickTime(clusterCfg.getTickTime());
				quorumPeer.setMinSessionTimeout(clusterCfg
						.getMinSessionTimeout());
				quorumPeer.setMaxSessionTimeout(clusterCfg
						.getMaxSessionTimeout());
				quorumPeer.setInitLimit(clusterCfg.getInitLimit());
				quorumPeer.setSyncLimit(clusterCfg.getSyncLimit());
				quorumPeer.setQuorumVerifier(quorumVerifier);
				quorumPeer.setCnxnFactory(cnxnFactory);
				quorumPeer.setZKDatabase(new ZKDatabase(quorumPeer
						.getTxnFactory()));
				quorumPeer.setLearnerType(getLearnerType(clusterCfg
						.getPeerType()));

				quorumPeer.start();
				quorumPeer.join();
			}

		} catch (Exception e) {
			LOG.error(e.getMessage(), e);
		}

	}

	protected LearnerType getLearnerType(String type) {
		return LearnerType.PARTICIPANT.name().equalsIgnoreCase(type) ? LearnerType.PARTICIPANT
				: LearnerType.OBSERVER;
	}

	protected File dirExists(String dir) throws IOException {
		File file = new File(dir);
		if (!file.isDirectory())
			throw new IOException("dataDir is not directory.");
		if (!file.exists()) {
			file.createNewFile();
		}
		return file;
	}

	protected Long readMyId(File file) throws IOException {
		if (file == null || !file.exists())
			return null;
		BufferedReader br = new BufferedReader(new FileReader(file));
		try {
			String value = br.readLine();
			if (value != null)
				return Long.parseLong(value);
			return null;
		} finally {
			br.close();
		}
	}

	protected File fileExists(File dir, String name) throws IOException {
		File file = new File(dir, name);
		if (!file.exists()) {
			file.createNewFile();
		}
		return file;
	}

	/**
	 * @param args
	 */
	public static void main(String[] args) {
		Options opts = new Options();
		opts.addOption(Constants.HELP_ARG, false,
				"Print help for this application");
		opts.addOption(Constants.FILE_ARG, true, "config file");
		opts.addOption(Constants.PORT_ARG, true, "client port");
		opts.addOption(Constants.DATA_DIR_ARG, true, "data directory");
		BasicParser parser = new BasicParser();
		try {
			CommandLine cl = parser.parse(opts, args);

			if (cl.hasOption(Constants.HELP_ARG)) {
				HelpFormatter hf = new HelpFormatter();
				hf.printHelp("arguments", opts);
			} else {
				String config = cl.getOptionValue(Constants.FILE_ARG);
				Integer port = cl.hasOption(Constants.PORT_ARG) ? Integer
						.valueOf(cl.getOptionValue(Constants.PORT_ARG)) : null;
				String dataDir = cl.getOptionValue(Constants.DATA_DIR_ARG);
				ClusterServer cluster = new ClusterServer(config, port, dataDir);
				cluster.run();
			}
		} catch (ParseException e) {
			LOG.error(e.getMessage(), e);
		} catch (Exception e) {
			LOG.error(e.getMessage(), e);
		}

	}
}
